package com.example.javarabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * RabbitMQ生产者
 * @author HAN Bing
 * @date 2021/5/29
 */
public class RabbitProducer {
    private static final String EXCHANGE_NAME = "exchange_demo";
    private static final String ROUTING_KEY = "routingkey_demo";
    private static final String QUEUE_NAME = "queue_demo";
    private static final String IP_ADDRESS = "8.131.233.42";
    private static final int PORT = 5672;

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername("root");
        factory.setPassword("d");

        // 创建连接
        Connection connection = factory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();

        // 创建一个tpye为direct，持久化的（将交换器写入硬盘，重启服务器不会丢失），
        // 非自动删除的（若为true，交换器先前绑定的队列或交换器都与此解绑时，交换器被自动删除），
        // 非内置（内置是指客户端程序无法发送消息到这个交换器，只能通过交换器路由到交换器这种方式）的交换器
        //                       交换器名称           类型       持久化    非自动删除     非内置   其他结构化参数
        channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, false, null);

        // 创建一个持久化，非排他的（不同connection可共用），非自动删除（连接断开时不自动删除）的队列
        //                    队列名称      持久化     非排他    非自动删除    其他参数
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 将交换器与队列通过路由键绑定
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

        // 发送一条持久化的消息
        String message = "Hellow World!";
        // mandatory，找不到队列，为true时返回给生产者，为false时销毁
        // immediate，在3.0版本后废弃，用TTL和DLX。如果队列中没有消费者，则将消息返还给生产者。
        //                      交换器         路由键        非强制       非immediate             消息基本属性（包含14个）             消息体（payload）
        channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, false,     false,    MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());


//        // 捕获mandatory为true时返回的消息
//        channel.addReturnListener(new ReturnListener() {
//            public void handleReturn(int replyCode, String replyText,
//                                     String exchange, String routingKey,
//                                     AMQP.BasicProperties basicProperties,
//                                     byte[] body) throws IOException {
//                String message = new String(body);
//                System.out.println("Basic.Return返回的结果是：" + message);
//            }
//        });


//        // 设置备份交换器，无需编写以上捕获类即可捕获mandatory退回的消息。
//        Map<String, Object> args1 = new HashMap<>();
//        args1.put("alternate-exchange", "myAe");
//        // 把myAe指定成normalExchange的备份交换器
//        channel.exchangeDeclare("normalExchange", "direct", true, false, args1);
//        // 定义myAe交换器，一般用fanout
//        channel.exchangeDeclare("myAe", "fanout", true, false, null);
//
//        channel.queueDeclare("normalQueue", true, false, false, null);
//        channel.queueBind("normalQueue", "normalExchange", "normalKey");
//        channel.queueDeclare("unroutedQueue", true, false, false, null);
//        channel.queueBind("unroutedQueue", "myAe", "normalKey");


//        // 设置过期时间TTL
//        // 第一种，队列中所有消息的TTL，一旦消息过期，就从队列中抹去
//        Map<String, Object> args2 = new HashMap<>();
//        // 毫秒
//        args2.put("x-message-ttl", 6000);
//        channel.queueDeclare(queueName, durable, exclusive, autoDelete, args2);
//
//        // 第二种，每条消息的TTL，即使消息过期，也不会马上抹去，等到即将被消费之前判定
//        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
//        // 持久化
//        builder.deliveryMode(2);
//        builder.expiration("6000");
//        AMQP.BasicProperties properties = builder.build();
//        channel.basicPublish(exchangeName, routingKey, mandatory, properties, "myTTLMessage".getBytes());
//
//        // 第三种，队列的TTL
//        Map<String, Object> args3 = new HashMap<>();
//        args3.put("x-expires", 6000);
//        channel.queueDeclare("myQueue", false, false, false, args3);


//        // 死信队列
//        // 声明死信交换器
//        channel.exchangeDeclare("exchange.dlx", "direct", true);
//        // 声明正常交换器
//        channel.exchangeDeclare("exchange.normal", "fanout", true);
//
//        // 构造关于死信的参数
//        Map<String, Object> args4 = new HashMap<>();
//        args4.put("x-message-ttl", 10000);
//        args4.put("x-dead-letter-exchange", "exchange.dlx");
//        args4.put("x-dead-letter-routing-key", "routingkey"); // 可省略， 如果省略，则使用正常队列的routingkey
//
//        // 声明正常队列，关于死信的参数放在正常队列中
//        channel.queueDeclare("queue.normal", true, false, false, args4);
//        // 绑定正常交换器和正常队列，由于是fanout，routingkey设置为空字符串
//        channel.queueBind("queue.normal", "exchange.normal", "");
//
//        // 声明队列，绑定死信交换器，这个队列就成了死信队列
//        channel.queueDeclare("queue.dlx", true, false, false, null);
//        channel.queueBind("queue.dlx", "exchange.dlx", "routingkey"); // 因为DLX类型是direct，所以此处路由键"routingkey"与args4中一致
//
//        // 发测试消息，消息会在10秒后过期，进入DLX，然后通过路由键"routingkey"匹配进死信队列
//        // 由此可实现延迟队列，即消费者过一段时间才消费消息，消费的是死信队列里的消息
//        channel.basicPublish("exchange.normal", "rk", MessageProperties.PERSISTENT_TEXT_PLAIN, "dlx".getBytes());


//        // 消息优先级
//        // 首先，设置队列的最大优先级
//        Map<String, Object> args5 = new HashMap<>();
//        args5.put("x-max-priority", 10);
//        channel.queueDeclare("queue.priority", true, false, false, args5);
//        // 然后，设置消息的优先级，不能超过队列的最大优先级。如果消息总是马上被消费，则设置优先级没有意义。
//        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
//        builder.priority(5);
//        AMQP.BasicProperties properties = builder.build();
//        channel.basicPublish("exchange_priority", "rk_priority", properties, "messages".getBytes());


//        // 生产者确认：
//        // 第一种：事务机制
//        // 第一步，将信道设置为事务模式
//        channel.txSelect();
//        for (int i = 0; i < 20; i++) {
//            try {
//                channel.basicPublish(exchange, routingkey, MessageProperties.PERSISTENT_TEXT_PLAIN, "msg".getBytes());
//                // 第二步，提交事务，只有消息成功被RabbitMQ接收，事务才能提交成功
//                channel.txCommit();
//            } catch (IOException e) {
//                e.printStackTrace();
//                // 如果事务没有提交成功，第三步，回滚事务
//                channel.txRollback();
//            }
//        }
//        // 第二种：发送方确认。另有批量确认（电子书1383）和异步确认（电子书1392）可以提高吞吐量
//        try{
//            // 将信道设置为publisher confirm模式。一旦消息被投递到所有匹配的队列之后，RabbitMQ就会发送一个Basic.Ack给生产者
//            channel.confirmSelect();
//            channel.basicPublish("exchange", "routingKey", null, "msg".getBytes());
//            // 如果生产者收到了Basic.Ack，则返回true
//            if (!channel.waitForConfirms()) {
//                System.out.println("send message failed");
//                // do something else
//            }
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }



        // 关闭资源
        channel.close();
        connection.close();
    }

}
